package com.gome.ocean.server.oracle.common.db.rocketmq;

import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.gome.ocean.server.oracle.common.lifecycle.AbstractYuGongLifeCycle;
import com.gome.ocean.server.oracle.common.lifecycle.YuGongLifeCycle;
import com.gome.ocean.server.oracle.exception.YuGongException;
import com.google.common.base.Function;
import com.google.common.collect.MigrateMap;

/**
 * RocketMq工厂
 * 
 * @author haqiaolong 2016年6月27日 下午6:00:06
 * @since 5.0.0
 */
public class RocketMqSourceFactory extends AbstractYuGongLifeCycle implements YuGongLifeCycle {

    private static final Logger                     logger = LoggerFactory.getLogger(RocketMqSourceFactory.class);

    private Map<RocketMqConfig, RocketMqDataSource> rocketMqDatasources;

    public void start() {
        super.start();
        rocketMqDatasources = MigrateMap.makeComputingMap(new Function<RocketMqConfig, RocketMqDataSource>() {

            public RocketMqDataSource apply(RocketMqConfig config) {
                DefaultMQProducer producer = new DefaultMQProducer("p".concat(config.getTopicName()));
                producer.setNamesrvAddr(config.getNamesrvAddr());
                try {
                    producer.start();
                } catch (MQClientException e) {
                    logger.error("create mqdataSource error!", e);
                    throw new YuGongException("create mqdataSource error!", e);
                }
                return new RocketMqDataSource(producer, config);
            }
        });

    }

    public void stop() {
        super.stop();

        for (RocketMqDataSource rocketMqDataSource : rocketMqDatasources.values()) {
            rocketMqDataSource.getProducer().shutdown();
        }
        rocketMqDatasources.clear();
    }

    public RocketMqDataSource getDataSource(RocketMqConfig config) {
        return rocketMqDatasources.get(config);
    }
}
